#!/bin/bash

IP=`/sbin/ifconfig -a|grep inet|grep -v 127.0.0.1|grep -v 172.17.0.1|grep -v 10.0.2.15|grep -v inet6|awk '{print $2}'|tr -d "addr:"`

# 删除数据目录及窗器
rm -rf /opt/module/es/logstash

# 下载logstash安装包
URL=https://artifacts.elastic.co/downloads/logstash/logstash-6.5.4.tar.gz
source ../linux/download_unzip.sh "es/logstash" $URL

# 下载mysql连接驱动包
URL=https://repo1.maven.org/maven2/mysql/mysql-connector-java/5.1.48/mysql-connector-java-5.1.48.jar
source ../linux/download_unzip.sh "es/logstash" $URL
#============== 调用本脚本时可用变量 ==============
#>>> $FILE_PATH 文件绝对路径：/opt/software/logstash-6.5.4.tar.gz
#>>> $FILE_EXTENSION 文件后缀：gz
#>>> $File_HOME 软件HOME：/opt/module/es/logstash/logstash-6.5.4
#>>> $REPO_PATH 仓库目录：/opt/repository/es/logstash
#============== 调用本脚本时可用变量 ==============
#>>> $FILE_PATH 文件绝对路径：/opt/software/mysql-connector-java-5.1.48.jar
#>>> $FILE_EXTENSION 文件后缀：jar
#>>> $File_HOME 软件HOME：/opt/module/es/logstash/logstash-6.5.4
#>>> $REPO_PATH 仓库目录：/opt/repository/es/logstash



mkdir -p $File_HOME/mysql
# (1)mysql驱动包复制到挂载目录
cp $FILE_PATH $File_HOME/mysql

# 安装插件
echo ">>> 开始安装：mysql输入 es输出插件"
$File_HOME/bin/logstash-plugin install logstash-input-jdbc
$File_HOME/bin/logstash-plugin install logstash-output-elasticsearch

echo ">>> 开始配置：mysql连接 es输出地址"
# (3)管连接配置（多库，多表时，复制一份jdbc{}，其中type改名）
cat >$File_HOME/mysql/mysql.config<<EOF
input {
	stdin { }
	jdbc {
		type => "jdbc"
		jdbc_connection_string => "jdbc:mysql://$IP/test?characterEncoding=UTF-8&autoReconnect=true"
		jdbc_user => "root"
		jdbc_password => "123456"
		jdbc_driver_library => "$File_HOME/mysql/mysql-connector-java-5.1.48.jar"
		jdbc_driver_class => "com.mysql.jdbc.Driver"
		connection_retry_attempts => "3"
		jdbc_validate_connection => "true"
		jdbc_validation_timeout => "3600"
		jdbc_paging_enabled => "true"
		jdbc_page_size => "500"
		statement => "SELECT id,name FROM test WHERE id > :sql_last_value"
		lowercase_column_names => false
		sql_log_level => info
		record_last_run => true
		use_column_value => true
		tracking_column => "id"
		tracking_column_type => numeric
		last_run_metadata_path => "$File_HOME/mysql/last_id.txt"
		clean_run => false
		schedule => "* * * * *"
	}
}
output {
    if[type] == "jdbc" {
        stdout {
            codec => json_lines
        }
        elasticsearch {
            hosts => ["$IP:9200"]
            index => "test"
            document_type => "test"
            document_id => "%{id}"
            template => "$File_HOME/mysql/test.json"
            template_name => "test-*"
            template_overwrite => true
        }
    }
}
EOF

cat >$File_HOME/mysql/test.json<<EOF
{
    "template":"test-*",
    "settings": {
        "number_of_replicas": "1",
        "number_of_shards": "3",
        "index": {
            "refresh_interval": "10s"
        }
    },
    "mappings":{
        "test":{
            "properties":{
                "id":{
                    "type":"integer"
                },
                "name":{
                    "type": "text",
                    "analyzer": "ik_smart",
                    "search_analyzer": "ik_smart"
                }
            }
        }
    }
}
EOF

echo "启动命令：$File_HOME/bin/logstash -f $File_HOME/mysql/mysql.config"
echo "启动命令：nohup $File_HOME/bin/logstash -f $File_HOME/mysql/mysql.config >/dev/null &"

#=============================================================
#                          配置说明
#=============================================================
# input {
#     stdin { }
#     jdbc {
#         type => "jdbc"
#             # 数据库连接地址
#         jdbc_connection_string => "jdbc:mysql://$IP/test?characterEncoding=UTF-8&autoReconnect=true""
#             # 数据库连接账号密码；
#         jdbc_user => "root"
#         jdbc_password => "123456"
#             # MySQL依赖包路径；
#         jdbc_driver_library => "/usr/share/logstash/mysql/mysql-connector-java-5.1.48.jar"
#             # the name of the driver class for mysql
#         jdbc_driver_class => "com.mysql.jdbc.Driver"
#             # 数据库重连尝试次数
#         connection_retry_attempts => "3"
#             # 判断数据库连接是否可用，默认false不开启
#         jdbc_validate_connection => "true"
#             # 数据库连接可用校验超时时间，默认3600S
#         jdbc_validation_timeout => "3600"
#             # 开启分页查询（默认false不开启）；
#         jdbc_paging_enabled => "true"
#             # 单次分页查询条数（默认100000,若字段较多且更新频率较高，建议调低此值）；
#         jdbc_page_size => "500"
#             # statement为查询数据sql，如果sql较复杂，建议配通过statement_filepath配置sql文件的存放路径；
#             # sql_last_value为内置的变量，存放上次查询结果中最后一条数据tracking_column的值，此处即为ModifyTime；
#             # statement_filepath => "/usr/share/logstash/mysql/jdbc.sql"
#         statement => "SELECT id,name FROM 'mytable' WHERE id>= :sql_last_value"
#             # 是否将字段名转换为小写，默认true（如果有数据序列化、反序列化需求，建议改为false）；
#         lowercase_column_names => false
#             # Value can be any of: fatal,error,warn,info,debug，默认info；
#         sql_log_level => info
#             #
#             # 是否记录上次执行结果，true表示会将上次执行结果的tracking_column字段的值保存到last_run_metadata_path指定的文件中；
#         record_last_run => true
#             # 需要记录查询结果某字段的值时，此字段为true，否则默认tracking_column为timestamp的值；
#         use_column_value => true
#             # 需要记录的字段，用于增量同步，需是数据库字段
#         tracking_column => "id"
#             # Value can be any of: numeric,timestamp，Default value is "numeric"
#         tracking_column_type => numeric
#             # record_last_run上次数据存放位置；
#         last_run_metadata_path => "/usr/share/logstash/mysql/last_id.txt"
#             # 是否清除last_run_metadata_path的记录，需要增量同步时此字段必须为false；
#         clean_run => false
#             #
#             # 同步频率(分 时 天 月 年)，默认每分钟同步一次；
#         schedule => "* * * * *"
#     }
# }

# filter {
#     json {
#         source => "message"
#         remove_field => ["message"]
#     }
# }

# output {
#     stdout {
#         codec => json_lines
#     }
#     elasticsearch {
#         #注意mysql连接地址一定要用ip，不能使用localhost等
#         hosts => ["$IP:9200"]
#         index => "test"
#         document_type => "_doc"
#         document_id => "%{id}"
#     }
# }



